package com.wsjj.gmall;

import com.wsjj.gmall.base.BaseSQLApp;
import com.wsjj.gmall.constant.Constant;
import com.wsjj.gmall.util.SQLUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 *
 *下单事实表
 *
 *
 */
public class DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(10014,4,"wy");
    }


    @Override
    public void handle(StreamTableEnvironment streamTable) {
//        TODO 1.设置状态过期时间-----------------不然会oom、
        streamTable.getConfig().setIdleStateRetention(Duration.ofSeconds(10L));
//        TODO 2.从kafka中读取数据
        readOdsTopicdb(streamTable);
//        TODO 3.读取的数据中筛选出下单方面的数据  order_detail(订单明细表)   order_info(订单表)  order_detail_activity(订单活动关联表)    order_detail_coupon(订单优惠卷关联表)    insert
//        order_detail
        Table order_detail_table = streamTable.sqlQuery("select `data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['sku_name'] sku_name,\n" +
                "`data`['order_price'] order_price,\n" +
                "`data`['sku_num'] sku_num,\n" +
                "`data`['split_total_amount'] split_total_amount,\n" +
                "`data`['split_activity_amount'] split_activity_amount,\n" +
                "`data`['split_coupon_amount'] split_coupon_amount,\n" +
                "ts\n" +
                "from KafkaTable\n" +
                "where type='insert' and `table`='order_detail'");


//        order_info
        Table order_info_table = streamTable.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['refundable_time'] refundable_time,\n" +
                "ts\n" +
                "from KafkaTable\n" +
                "where `table`='order_info' and type='insert'");

//        order_detail_activity
        Table order_detail_activity_table = streamTable.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['order_detail_id'] order_detail_id,\n" +
                "`data`['activity_id'] activity_id,\n" +
                "`data`['activity_rule_id'] activity_rule_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "       ts\n" +
                "from KafkaTable\n" +
                "where `table`='order_detail_activity' and type = 'insert';");


//        order_detail_activity_table.execute().print();

//        order_detail_coupon
        Table order_detail_coupon_table = streamTable.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['order_detail_id'] order_detail_id,\n" +
                "`data`['coupon_id'] coupon_id,\n" +
                "`data`['coupon_use_id'] coupon_use_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "       ts\n" +
                "from KafkaTable\n" +
                "where `table`='order_detail_coupon' and type = 'insert'");

//        order_detail_coupon_table.execute().print();


//        order_detail_table.execute().print();

//        order_info_table.execute().print();

//        order_detail_table.execute().print();

        streamTable.createTemporaryView("order_detail_table",order_detail_table);
        streamTable.createTemporaryView("order_info_table",order_info_table);
        streamTable.createTemporaryView("order_detail_activity_table",order_detail_activity_table);
        streamTable.createTemporaryView("order_detail_coupon_table",order_detail_coupon_table);
//        TODO 4.关联4张表的数据
        Table table = streamTable.sqlQuery("select\n" +
                "       t1.id order_detail_id,\n" +
                "t1.order_id order_id,\n" +
                "t1.sku_id sku_id,\n" +
                "       sku_name sku_name,\n" +
                "       order_price order_price,\n" +
                "       sku_num ,\n" +
                "       split_total_amount,\n" +
                "       split_activity_amount,\n" +
                "       split_coupon_amount,\n" +
                "       consignee,\n" +
                "       consignee_tel,\n" +
                "       total_amount,\n" +
                "       order_status,\n" +
                "       user_id,\n" +
                "       payment_way,\n" +
                "       province_id,\n" +
                "       activity_reduce_amount,\n" +
                "       coupon_reduce_amount,\n" +
                "       original_total_amount,\n" +
                "       refundable_time,\n" +
                "       activity_id,\n" +
                "       activity_rule_id,\n" +
                "       coupon_id,\n" +
                "       coupon_use_id,\n" +
                "       t1.ts ts\n" +
                "from order_detail_table t1\n" +
                "join  order_info_table t2\n" +
                "on t1.order_id = t2.id\n" +
                "left join order_detail_activity_table t3\n" +
                "on t1.id = t3.order_detail_id\n" +
                "left join order_detail_coupon_table t4\n" +
                "on t1.id = t4.order_detail_id");
//        table.execute().print();


//        TODO 5.关联好的数据写入到kafka
        streamTable.executeSql("CREATE TABLE "+Constant.TOPIC_DWD_TRADE_ORDER_DETAIL+" (\n" +
                "order_detail_id STRING,\n" +
                "order_id STRING,\n" +
                "sku_id STRING,\n" +
                "sku_name string,\n" +
                "order_price string,\n" +
                "sku_num string,\n" +
                "split_total_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "consignee string,\n" +
                "consignee_tel string,\n" +
                "total_amount string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "payment_way string,\n" +
                "province_id string,\n" +
                "activity_reduce_amount string,\n" +
                "coupon_reduce_amount string,\n" +
                "original_total_amount string,\n" +
                "refundable_time string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "coupon_id string,\n" +
                "coupon_use_id string,\n" +
                "ts BIGINT,\n" +
                "PRIMARY KEY (order_detail_id) NOT ENFORCED\n" +
                ") " + SQLUtil.getUpdateKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));

        table.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);









    }
}
